package zhanghao;




import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat;
import org.apache.hadoop.hbase.mapreduce.KeyValueSortReducer;
import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;

import com.bj58.analysis.track.format.TrackInputFormat;


public class LCSourceFlowDriver  extends Configured implements Tool{
	private static Logger LOG = Logger.getLogger(LCSourceFlowDriver.class);
	public int run(String[] args) throws Exception {

		Configuration conf = getConf();
		//-D   hbase.zookeeper.quorum=10.126.81.131,10.126.81.132,10.126.81.133,10.126.81.221,10.126.81.222
		
		
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

		if (otherArgs.length != 7) {
			LOG.info("************************************************");
			LOG.info("************************************************");
			LOG.info("Usage: please input 1 params, for example: file.jar args[0]");
			LOG.info("args[0] is dateList: 20130101,20130106 or 20130106");
			LOG.info("args[1] is input_pc: /home/hdp_58_common/resultdata/all_site_user_action_etl/pc.track.58.com");
			LOG.info("args[2] is input_m: /home/hdp_58_common/resultdata/all_site_user_action_etl/m.track.58.com");
			//  /home/hdp_58_common/resultdata/all_site_user_action_etl/m.track.chinahr.com
			//  /home/hdp_58_common/resultdata/all_site_user_action_etl/pc.track.chinahr.com
			LOG.info("args[3] is input_chinahr_pc: /home/hdp_58_common/resultdata/all_site_user_action_etl/pc.track.chinahr.com");
			LOG.info("args[4] is input_chinahr_m: /home/hdp_58_common/resultdata/all_site_user_action_etl/m.track.chinahr.com");			
			LOG.info("args[5] is output_base(结果路径前缀): /home/hdp_lbg_ecdata_basestat/middata/liucheng01/sourceflow/result");
			LOG.info("args[6] is output_hbase(存储): /home/hbase_bulkimport/hdp_lbg_ecdata_basestat/liucheng01/sourceall");
			LOG.info("************************************************");
			LOG.info("************************************************");
			System.exit(127);
		}
		

		String startDate = otherArgs[0].split(",")[0];
		String endDate 	 = otherArgs[0].split(",").length == 2 ? otherArgs[0].split(",")[1] : startDate;
		String baseInPath_track_pc = otherArgs[1];
		String baseInPath_track_m = otherArgs[2];
		String baseInPath_track_chinahr_pc = otherArgs[3];
		String baseInPath_track_chinahr_m = otherArgs[4];
		LOG.info("startDate = " + startDate);
		LOG.info("baseInPath_track_pc = " + baseInPath_track_pc);
		LOG.info("baseInPath_track_m = " + baseInPath_track_m);
		LOG.info("baseInPath_track_chinahr_pc = " + baseInPath_track_chinahr_pc);
		LOG.info("baseInPath_track_chinahr_m = " + baseInPath_track_chinahr_m);
		
		
		String baseOutPath_PC = otherArgs[5]+"/pc";
		String baseOutPath_M = otherArgs[5]+"/m";
		String baseOutPath_Chinahr_PC = otherArgs[5]+"/chinahr_pc";
		String baseOutPath_Chinahr_M = otherArgs[5]+"/chinahr_m";
		
		String baseOutHbase = otherArgs[6];
		
		LOG.info("baseOutPath_PC = " + baseOutPath_PC);
		LOG.info("baseOutPath_M = " + baseOutPath_M);
		LOG.info("baseOutPath_Chinahr_PC = " + baseOutPath_Chinahr_PC);
		LOG.info("baseOutPath_Chinahr_M = " + baseOutPath_Chinahr_M);
		LOG.info("baseOutHbase = " + baseOutHbase);
		
		
		
		int returnCode = 126;
		
		
		
		
		for (String runDate : DateUtil.getDateList(startDate, endDate))
		{
			LOG.info("*****   新一轮执行开始    *****");
			
			conf.set("DATE", runDate);
			conf.set("PLATFORM", "PC");
			Job job_PC = Job.getInstance(conf);
			job_PC.setJobName("LCSourceflowDriver_PC");
			job_PC.setJarByClass(LCSourceFlowDriver.class);
			job_PC.setInputFormatClass(TrackInputFormat.class);
			job_PC.setMapperClass(SourceAllFlowlMap.class);
			job_PC.setReducerClass(SourceAllFlowReduce.class);
			job_PC.setNumReduceTasks(30);
			job_PC.setOutputKeyClass(Text.class);
			job_PC.setOutputValueClass(Text.class);
			job_PC.setMapOutputKeyClass(Text.class);
			job_PC.setMapOutputValueClass(Text.class);
			
			if(FileSystem.get(conf).exists(new Path(baseInPath_track_pc + "/"+ runDate)))
			{
				FileInputFormat.addInputPath(job_PC, new Path(baseInPath_track_pc + "/"+ runDate));
			}
			
			
			FileSystem.get(conf).delete(new Path(baseOutPath_PC  + "/" + runDate), true);
			FileOutputFormat.setOutputPath(job_PC, new Path(baseOutPath_PC  + "/" + runDate));
			returnCode = job_PC.waitForCompletion(true) ? 0 : 1;
			LOG.info("********************************job_PC " + runDate + " *********************");
			if (returnCode != 0){
				return returnCode;
			}
				
			LOG.info("*****   job_PC 运行结束   *****");
			
			
			
			conf.set("PLATFORM", "M");
			Job job_M = Job.getInstance(conf);
			job_M.setJobName("LCSourceflowDriver_M");
			job_M.setJarByClass(LCSourceFlowDriver.class);
			job_M.setInputFormatClass(TrackInputFormat.class);
			job_M.setMapperClass(SourceAllFlowlMap.class);
			job_M.setReducerClass(SourceAllFlowReduce.class);
			job_M.setNumReduceTasks(30);
			job_M.setOutputKeyClass(Text.class);
			job_M.setOutputValueClass(Text.class);
			job_M.setMapOutputKeyClass(Text.class);
			job_M.setMapOutputValueClass(Text.class);
			
			if( FileSystem.get(conf).exists(new Path(baseInPath_track_m + "/"+ runDate)))
			{
				FileInputFormat.addInputPath(job_M, new Path(baseInPath_track_m + "/"+ runDate));
			}
			
			FileSystem.get(conf).delete(new Path(baseOutPath_M  + "/" + runDate), true);
			FileOutputFormat.setOutputPath(job_M, new Path(baseOutPath_M  + "/" + runDate));
			returnCode = job_M.waitForCompletion(true) ? 0 : 1;
			LOG.info("********************************job_M " + runDate + " *********************");
			if (returnCode != 0){
				return returnCode;
			}
				
			LOG.info("*****   job_M 运行结束   *****");
			
			
			
			
			
			conf.set("PLATFORM", "Chinahr_PC");
			Job job_Chinahr_PC = Job.getInstance(conf);
			job_Chinahr_PC.setJobName("LCSourceflowDriver_Chinahr_PC");
			job_Chinahr_PC.setJarByClass(LCSourceFlowDriver.class);
			job_Chinahr_PC.setInputFormatClass(TrackInputFormat.class);
			job_Chinahr_PC.setMapperClass(SourceAllFlowlMap.class);
			job_Chinahr_PC.setReducerClass(SourceAllFlowReduce.class);
			job_Chinahr_PC.setNumReduceTasks(30);
			job_Chinahr_PC.setOutputKeyClass(Text.class);
			job_Chinahr_PC.setOutputValueClass(Text.class);
			job_Chinahr_PC.setMapOutputKeyClass(Text.class);
			job_Chinahr_PC.setMapOutputValueClass(Text.class);
			
			if( FileSystem.get(conf).exists(new Path(baseInPath_track_chinahr_pc + "/"+ runDate)))
			{
				FileInputFormat.addInputPath(job_Chinahr_PC, new Path(baseInPath_track_chinahr_pc + "/"+ runDate));
			}
			
			
			FileSystem.get(conf).delete(new Path(baseOutPath_Chinahr_PC  + "/" + runDate), true);
			FileOutputFormat.setOutputPath(job_Chinahr_PC, new Path(baseOutPath_Chinahr_PC  + "/" + runDate));
			returnCode = job_Chinahr_PC.waitForCompletion(true) ? 0 : 1;
			LOG.info("********************************job_Chinahr_PC " + runDate + " *********************");
			if (returnCode != 0){
				return returnCode;
			}
				
			LOG.info("*****   job_Chinahr_PC 运行结束   *****");
			
			
			
			conf.set("PLATFORM", "Chinahr_M");
			Job job_Chinahr_M = Job.getInstance(conf);
			job_Chinahr_M.setJobName("LCSourceflowDriver_Chinahr_M");
			job_Chinahr_M.setJarByClass(LCSourceFlowDriver.class);
			job_Chinahr_M.setInputFormatClass(TrackInputFormat.class);
			job_Chinahr_M.setMapperClass(SourceAllFlowlMap.class);
			job_Chinahr_M.setReducerClass(SourceAllFlowReduce.class);
			job_Chinahr_M.setNumReduceTasks(30);
			job_Chinahr_M.setOutputKeyClass(Text.class);
			job_Chinahr_M.setOutputValueClass(Text.class);
			job_Chinahr_M.setMapOutputKeyClass(Text.class);
			job_Chinahr_M.setMapOutputValueClass(Text.class);
			
			if(FileSystem.get(conf).exists(new Path(baseInPath_track_chinahr_m + "/"+ runDate)))
			{
				FileInputFormat.addInputPath(job_Chinahr_M, new Path(baseInPath_track_chinahr_m + "/"+ runDate));
			}

			
			FileSystem.get(conf).delete(new Path(baseOutPath_Chinahr_M  + "/" + runDate), true);
			FileOutputFormat.setOutputPath(job_Chinahr_M, new Path(baseOutPath_Chinahr_M  + "/" + runDate));
			returnCode = job_Chinahr_M.waitForCompletion(true) ? 0 : 1;
			LOG.info("********************************job_Chinahr_M " + runDate + " *********************");
			if (returnCode != 0){
				return returnCode;
			}
				
			LOG.info("*****   job_Chinahr_M 运行结束   *****");
			
			
			
			
			
			// hbase入库job
			LOG.info("*****   job_hbase 开始运行   *****");
			Job convertOutputToHFileJob = Job.getInstance(conf);
			convertOutputToHFileJob.setJobName("LCSourceFlow_bulkload");
			convertOutputToHFileJob.setJarByClass(LCSourceFlowDriver.class);
			convertOutputToHFileJob.setMapperClass(SourceAllFlowHbaseMap.ConvertOutToHFileMapper.class);
			convertOutputToHFileJob.setReducerClass(KeyValueSortReducer.class);
			
			convertOutputToHFileJob.setMapOutputKeyClass(ImmutableBytesWritable.class);
			convertOutputToHFileJob.setMapOutputValueClass(KeyValue.class);
			
			//以第一个Job的输出做为第二个Job的输入
			if(FileSystem.get(conf).exists(new Path(baseOutPath_PC + "/"+ runDate)))
			{
				FileInputFormat.addInputPath(convertOutputToHFileJob, new Path(baseOutPath_PC + "/"+ runDate));
			}
			if( FileSystem.get(conf).exists(new Path(baseOutPath_M + "/"+ runDate)))
			{
				FileInputFormat.addInputPath(convertOutputToHFileJob, new Path(baseOutPath_M + "/"+ runDate));
			}	
			if(FileSystem.get(conf).exists(new Path(baseOutPath_Chinahr_PC + "/"+ runDate)))
			{
				FileInputFormat.addInputPath(convertOutputToHFileJob, new Path(baseOutPath_Chinahr_PC + "/"+ runDate));
			}
			if( FileSystem.get(conf).exists(new Path(baseOutPath_Chinahr_M + "/"+ runDate)))
			{
				FileInputFormat.addInputPath(convertOutputToHFileJob, new Path(baseOutPath_Chinahr_M + "/"+ runDate));
			}	
			
			FileSystem.get(conf).delete(new Path(baseOutHbase+"/"+runDate), true);
			FileOutputFormat.setOutputPath(convertOutputToHFileJob, new Path(baseOutHbase+"/"+runDate));
			//创建HBase的配置对象
			Configuration hbaseConfiguration=HBaseConfiguration.create(conf);
			// 临时方案，hbase新平台参数
			
			
		    //创建目标表对象
//			HTable wordCountTable =new HTable(hbaseConfiguration, "LCHbaseTest");
		    HTable wordCountTable =new HTable(hbaseConfiguration, "LCSourceALL");
		    HFileOutputFormat.configureIncrementalLoad(convertOutputToHFileJob,wordCountTable);
		    
		    //提交第二个job
		    returnCode=convertOutputToHFileJob.waitForCompletion(true)?0:1;
		    //当第二个job结束之后，调用BulkLoad方式来将MR结果批量入库
		    LoadIncrementalHFiles loader = new LoadIncrementalHFiles(hbaseConfiguration);
		    //第一个参数为第二个Job的输出目录即保存HFile的目录，第二个参数为目标表
		    loader.doBulkLoad(new Path(baseOutHbase+"/"+runDate), wordCountTable);
		    LOG.info("************   job_hbase  结束" + runDate + " *************");
		}
		return returnCode;
	}
	
	
	public static void main(String[] args) throws Exception{
	    int status = ToolRunner.run(new LCSourceFlowDriver(), args);
	    System.exit(status);
	}
}
